package flink.clickhouse;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Random;
import java.util.UUID;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 使用jdbc-connector往clickhouse中写入数据
 */
public class clickhouseJDBCConnectorDemo {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT, "8081-8089");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setParallelism(1);

        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env, settings);


        DataStreamSource<Student> studentDataStreamSource = env.addSource(new SourceFunction<Student>() {
            private boolean flag = true;

            @Override
            public void run(SourceContext<Student> ctx) throws Exception {
                Random random = new Random();
                int id = 4000;
                while (id < 5000) {
                    id++;
                    Thread.sleep(100);
                    String name = UUID.randomUUID().toString();
                    int age = random.nextInt(100);
                    ctx.collect(new Student(id, name, age));
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        });

        studentDataStreamSource.print();

        tenv.createTemporaryView("source_student",studentDataStreamSource,$("id"),$("name"),$("age"));

        String querySql = "select cast(id as INTEGER) id,name,cast(age as TINYINT) age from source_student";
        Table etlResult = tenv.sqlQuery(querySql);
        DataStream<Row> rowDataStream = tenv.toAppendStream(etlResult, Row.class);
        rowDataStream.print();

        String sql = "CREATE TABLE sink_student (\n" +
                "     id INTEGER,\n" +
                "     name STRING,\n" +
                "     age TINYINT\n" +
                " ) WITH (\n" +
                "     'connector' = 'clickhouse',\n" +
                "     'url' = 'clickhouse://192.168.62.35:8123',\n" +
                "     'table-name' = 'student',\n" +
                "     'username' = 'liujian',\n" +
                "     'password' = '',\n"+
                "     'sink.batch-size' = '10',\n" +
                "     'sink.partition-strategy' = 'hash',\n" +
                "     'sink.partition-key' = 'id'\n" +
                " )";






        TableResult outputTable = tenv.executeSql(sql);

        tenv.executeSql("insert into sink_student select * from "+ etlResult);

        env.execute("");

    }


    public static class Student {
        private Integer id;
        private String name;
        private Integer age;

        public Student() {
        }

        public Student(Integer id, String name, Integer age) {
            this.id = id;
            this.name = name;
            this.age = age;
        }

        public Integer getId() {
            return id;
        }

        public void setId(Integer id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public Integer getAge() {
            return age;
        }

        public void setAge(Integer age) {
            this.age = age;
        }

        @Override
        public String toString() {
            return JSON.toJSONString(this);
        }
    }
}
